# -*- coding: utf-8 -*-
from datetime import timedelta
from time import time

from jms.dm.dm_transit_violation_sum_day_dt import jms_dm__dm_transit_violation_sum_day_dt
from utils.operators.rich_sql_sensor import RichSqlSensor


#doris表名
doris_table = "dm_transit_violation_sum_day_dt"
#label名
label = f"{doris_table}_{int(time())}"
#hive中表名
hive_table = "dm_transit_violation_sum_day_dt"
#hdfs文件路径
hdfs_path ="hdfs://{{var.value.hadoop_namespace}}/dw/hive/jms_dm.db/external/"
timeout = timedelta(hours=2).seconds
#判断当天分区是否存在
file_path = f'"{hdfs_path}{hive_table}/dt={{{{ execution_date | cst_ds }}}}/*"'
partition_lists = []
path_lists = []
for i in range(0, 7):
    if i == 0:
        partition_lists.append("p{{ execution_date | cst_ds_nodash }}")
        path_lists.append(file_path)
    else:
        partition_lists.append("p{{ execution_date | date_add(-" + str(i) + ") | cst_ds_nodash }}")
        path_lists.append(f'"{hdfs_path}{hive_table}/dt={{{{ execution_date | date_add(-{i}) | cst_ds }}}}/*"')

partition_str = ",".join(partition_lists)
path_str = ",".join(path_lists[::-1])
doris_jms_dm__dm_transit_violation_sum_day_dt = RichSqlSensor(
    task_id='doris_jms_dm__dm_transit_violation_sum_day_dt',
    pool='broker_load_pool',
    email=['jokic.wang@jtexpress.com'],
    task_concurrency=1,
    conn_id='doris',
    pre_sql=f"""
                 TRUNCATE TABLE jms_dm.{doris_table} PARTITION ({partition_str});
                 LOAD LABEL jms_dm.{label} (
                     DATA INFILE({path_str})
                     INTO TABLE {doris_table}
                     FORMAT AS 'PARQUET'(
                         statistic_date
                        ,wk_first_day
                        ,mn_first_day
                        ,trace_line
                        ,sender_center_code
                        ,sender_center_simple_name
                        ,sender_center_name
                        ,receiver_center_code
                        ,receiver_center_simple_name
                        ,receiver_center_name
                        ,center_code_1
                        ,center_simple_name_1
                        ,center_name_1
                        ,center_code_2
                        ,center_simple_name_2
                        ,center_name_2
                        ,center_code_3
                        ,center_simple_name_3
                        ,center_name_3
                        ,is_match_main
                        ,is_match_standby
                        ,route_name
                        ,standby_route_name
                        ,pre_mileage
                        ,main_next_code
                        ,main_next_name
                        ,main_next_next_code
                        ,main_next_next_name
                        ,standby_next_code
                        ,standby_next_name
                        ,standby_next_next_code
                        ,standby_next_next_name
                        ,real_all_line_code
                        ,real_all_line
                        ,dtype
                        ,is_over_volume
                        ,is_violation
                        ,transit_cnt
                        ,wrong_cnt
                        ,tmp_shipment_cnt
                        ,miss_cnt
                        ,dispatch_cnt
                        ,violate_cnt
                        ,package_total_volume
                        ,over_volume_trace_cnt
                        ,over_volume_transit_cnt
                        ,stop_shipment_cnt
                        ,no_route_cnt
                        ,bag_trace_err_cnt
                        ,bag_err_cnt
                        ,agent_code
                        ,agent_name
                        ,virt_code
                        ,virt_name
                        ,transit_rnk
                        ,over_volume_rnk
                        ,mn_violate_top_num
                        ,mn_over_volume_top_num
                        ,other_cnt
                        ,is_no_route
                        ,is_stop_shipment
                        ,in_province_volume_cnt
                        ,out_province_volume_cnt
                        ,airport_cnt
                        ,trace_warning_cnt
                 )
                 columns from path as (dt)
                 set(
                         statistic_date              =  statistic_date
                        ,dtype                       =  dtype
                        ,center_code_1               =  center_code_1
                        ,center_code_2               =  center_code_2
                        ,center_code_3               =  center_code_3
                        ,is_violation                =  is_violation
                        ,is_over_volume              =  is_over_volume
                        ,wk_first_day                =  wk_first_day
                        ,mn_first_day                =  mn_first_day
                        ,trace_line                  =  trace_line
                        ,sender_center_code          =  sender_center_code
                        ,sender_center_simple_name   =  sender_center_simple_name
                        ,sender_center_name          =  sender_center_name
                        ,receiver_center_code        =  receiver_center_code
                        ,receiver_center_simple_name =  receiver_center_simple_name
                        ,receiver_center_name        =  receiver_center_name
                        ,center_simple_name_1        =  center_simple_name_1
                        ,center_name_1               =  center_name_1
                        ,center_simple_name_2        =  center_simple_name_2
                        ,center_name_2               =  center_name_2
                        ,center_simple_name_3        =  center_simple_name_3
                        ,center_name_3               =  center_name_3
                        ,is_match_main               =  is_match_main
                        ,is_match_standby            =  is_match_standby
                        ,route_name                  =  route_name
                        ,standby_route_name          =  standby_route_name
                        ,pre_mileage                 =  pre_mileage
                        ,main_next_code              =  main_next_code
                        ,main_next_name              =  main_next_name
                        ,main_next_next_code         =  main_next_next_code
                        ,main_next_next_name         =  main_next_next_name
                        ,standby_next_code           =  standby_next_code
                        ,standby_next_name           =  standby_next_name
                        ,standby_next_next_code      =  standby_next_next_code
                        ,standby_next_next_name      =  standby_next_next_name
                        ,real_all_line_code          =  real_all_line_code
                        ,real_all_line               =  real_all_line
                        ,transit_cnt                 =  transit_cnt
                        ,wrong_cnt                   =  wrong_cnt
                        ,tmp_shipment_cnt            =  tmp_shipment_cnt
                        ,miss_cnt                    =  miss_cnt
                        ,dispatch_cnt                =  dispatch_cnt
                        ,violate_cnt                 =  violate_cnt
                        ,package_total_volume        =  package_total_volume
                        ,over_volume_trace_cnt       =  over_volume_trace_cnt
                        ,over_volume_transit_cnt     =  over_volume_transit_cnt
                        ,stop_shipment_cnt           =  stop_shipment_cnt
                        ,no_route_cnt                =  no_route_cnt
                        ,bag_trace_err_cnt           =  bag_trace_err_cnt
                        ,bag_err_cnt                 =  bag_err_cnt
                        ,agent_code                  =  agent_code
                        ,agent_name                  =  agent_name
                        ,virt_code                   =  virt_code
                        ,virt_name                   =  virt_name
                        ,transit_rnk                 =  transit_rnk
                        ,over_volume_rnk             =  over_volume_rnk
                        ,mn_violate_top_num          =  mn_violate_top_num
                        ,mn_over_volume_top_num      =  mn_over_volume_top_num
                        ,other_cnt                   =  other_cnt
                        ,is_no_route                 =  is_no_route
                        ,is_stop_shipment            =  is_stop_shipment
                        ,in_province_volume_cnt      =  in_province_volume_cnt
                        ,out_province_volume_cnt     =  out_province_volume_cnt
                        ,airport_cnt                 =  airport_cnt
                        ,trace_warning_cnt           =  trace_warning_cnt
                   )
                 )
                 WITH BROKER '{{{{ var.json.doris_brokers | random_choice }}}}'
                 PROPERTIES ('timeout'='{timeout}', 'max_filter_ratio'='0.0')""",
    poke_sql=f"SHOW LOAD FROM jms_dm WHERE label = '{label}' ORDER BY CreateTime DESC LIMIT 1",
    sql_on_kill=f"CANCEL LOAD FROM jms_dm WHERE LABEL = '{label}'",
    success=lambda r: r[2] == 'FINISHED',
    failure=lambda r: (r[2] is not None and r[2] == 'CANCELLED', str(r[7])),
    poke_interval=60,
    execution_timeout=timedelta(seconds=timeout + 120), )

doris_jms_dm__dm_transit_violation_sum_day_dt << [
    jms_dm__dm_transit_violation_sum_day_dt
]
